/**
 * Copyright © 2017 sdjictec.Co.Ltd. All rights reserved.
 **/
package com.sdjictec.haier;

import com.alibaba.fastjson.JSON;
import com.google.gson.Gson;
import com.sdjictec.haier.config.ProFileConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Arrays;
import java.util.Map;
import java.util.Properties;


public class KafkaToOracleTest {

    private static Logger log = LoggerFactory.getLogger(KafkaToOracleTest.class);

    public static void main(String[] args) {


//        String test = "1";


        Properties props = new Properties();
        props.put("bootstrap.servers", ProFileConfig.kafka_servers);
        props.put("group.id", "tttt111");
        props.put("enable.auto.commit", "false");
        props.put("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
        props.put("auto.offset.reset", "latest");

        KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props);
        consumer.subscribe(Arrays.asList(ProFileConfig.kafka_topic));

        Gson gson = new Gson();
        while (true) {
            ConsumerRecords<String, String> records = consumer.poll(100);

            if (records != null && !records.isEmpty()) {


                for (ConsumerRecord<String, String> record : records) {
                    try {
                        String json = record.value();


                        System.out.println("kafka 原始数据如下：");
                        System.out.println(json);
                        Map<String, Object> mapTypes = JSON.parseObject(json);
                        String uid = (String) mapTypes.get("uid");

                        consumer.commitAsync();
                    } catch (Exception e) {
                        consumer.commitAsync();
                        e.printStackTrace();
                        // System.out.println("数据插入失败！！！");
                        log.error("数据插入失败！！！");
                        log.error(record.value());
                    }
                }
            } else {
                //System.out.println("no data");
            }
        }

    }


}